﻿#pragma once
#include <iostream>
#include <string>
#include <future>
#include <thread>
#include <mutex>
#include "MyDeliveryReportCb.h"
using  namespace  std;

int kafka_thread() {
	std::string brokers = "127.0.0.1:9092";
	std::string topic = "test";
	std::string msg = "{'id':'11','name','koy'}";
	RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);

	std::string errstr;
	/* Set bootstrap broker(s) as a comma-separated list of
	* host or host:port (default port 9092).
	* librdkafka will use the bootstrap brokers to acquire the full
	* set of brokers from the cluster. */
	if (conf->set("bootstrap.servers", brokers, errstr) !=
		RdKafka::Conf::CONF_OK) {
		std::cerr << errstr << std::endl;
		exit(1);
	}

	MyDeliveryReportCb ex_dr_cb;

	if (conf->set("dr_cb", &ex_dr_cb, errstr) != RdKafka::Conf::CONF_OK) {
		std::cerr << errstr << std::endl;
		exit(1);
	}

	/*
	* Create producer instance.
	*/
	RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
	if (!producer) {
		std::cerr << "Failed to create producer: " << errstr << std::endl;
		exit(1);
	}
	delete conf;

	/*
	* Read messages from stdin and produce to broker.
	*/
	std::cout << "% Type message value and hit enter " <<
		"to produce message." << std::endl;

	std::string line = msg;

	if (line.empty()) {
		producer->poll(0);
		delete producer;
		return 0;
	}

retry:
	RdKafka::ErrorCode err =
		producer->produce(topic, RdKafka::Topic::PARTITION_UA,
		RdKafka::Producer::RK_MSG_COPY /* Copy payload */,
		/* Value */
		const_cast<char *>(line.c_str()), line.size(),
		/* Key */
		NULL, 0,
		/* Timestamp (defaults to current time) */
		0,
		/* Message headers, if any */
		NULL);

	if (err != RdKafka::ERR_NO_ERROR) {
		std::cerr << "% Failed to produce to topic " << topic << ": " <<
			RdKafka::err2str(err) << std::endl;

		if (err == RdKafka::ERR__QUEUE_FULL) {
			producer->poll(1000);/*block for max 1000ms*/
			goto retry;
		}

	}
	else {
		std::cerr << "% Enqueued message (" << line.size() << " bytes) " <<
			"for topic " << topic << std::endl;
	}

	producer->poll(0);
	std::cerr << "% Flushing final messages..." << std::endl;
	producer->flush(10 * 1000); /* wait for max 10 seconds */

	if (producer->outq_len() > 0)
		std::cerr << "% " << producer->outq_len() <<
		" message(s) were not delivered" << std::endl;

	while (true && producer->outq_len() > 0) {
		std::cerr << "Waiting for " << producer->outq_len() << std::endl;
		producer->poll(1000);
	}
	delete producer;

	return 0;
}

int main1() {
	cout << "main," << "thread_id" << std::this_thread::get_id << endl;
	std::future<int> res = std::async(kafka_thread);
	cout << "main,continue...." << endl;
	int def = 0;
	cout << "main,all over" << endl;
	return 0;

}

